package com.joizhang.reactor.framework;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.nio.channels.SelectionKey;
import java.util.concurrent.*;

/**
 * @author joizhang
 */
public class ThreadPoolDispatcher implements Dispatcher {

    private final ExecutorService executorService;

    public ThreadPoolDispatcher(int poolSize) {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("my-reactor-pool-%d").build();
        this.executorService = new ThreadPoolExecutor(
                poolSize,
                poolSize,
                120L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1024),
                threadFactory, new ThreadPoolExecutor.AbortPolicy());
    }

    @Override
    public void onChannelReadEvent(AbstractNioChannel channel, Object readObject, SelectionKey key) {
        executorService.execute(() -> channel.getHandler().handleChannelRead(channel, readObject, key));
    }

    @Override
    public void stop() throws InterruptedException {
        executorService.shutdown();
        executorService.awaitTermination(4, TimeUnit.SECONDS);
    }

}
